package com.ruoyi.system.mqtt;

import com.alibaba.fastjson.JSON;
import com.ruoyi.common.cache.CacheComponent;
import com.ruoyi.common.utils.SpringContextUtil;
import com.ruoyi.system.domain.ZsyyMqttConfig;
import com.ruoyi.system.mapper.ZsyyMqttConfigMapper;
import com.ruoyi.system.service.IZsyyMqttConfigService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @program: cloudpark
 * @description:
 * @author: cloud
 * @create: 2019-06-19 20:31
 **/
@Slf4j
public class MqttParkMessageCallBack implements MqttCallbackExtended {

    private SpringContextUtil springContextUtil = new SpringContextUtil();
    private CacheComponent cacheComponent =springContextUtil.getBean(CacheComponent.class);
    @Autowired
    public ZsyyMqttConfigMapper mqttConfigMapper ;

    @Override
    public void connectionLost(Throwable throwable) {
        String msg=throwable.getMessage() + "|" + ((MqttException) throwable).getReasonCode() + "|"+ throwable.getCause();
        MqttClient mqttClient = MqttParkClient.getMqttClient(MqttParkClient.MQTT_SERVER, MqttParkClient.MQTT_PORT,
                MqttParkClient.MQTT_CLIENTID);
        log.warn("MQPark##连接断开，当前clientID:{}，ErrorMsg:{}", mqttClient.getClientId(),msg);
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
       //接受到消息，进行处理
        String data = new String(mqttMessage.getPayload());
//        System.out.println("接受到消息data================"+data);
        Map map = JSON.parseObject(data,Map.class);
        String uuid = (String) map.get("uuid");
        cacheComponent.putRaw(uuid, data, 60*30);   //存入Redis30分钟
        String cmd = map.get("serviceId").toString().replace(".","A");
        try {
            Class<?> clazz = Class.forName("com.appapi.homecontroller.MqttDataController");
            Method method = clazz.getMethod(cmd,String.class);
            Object obj1 = method.invoke(springContextUtil.getBean(clazz),data);
        }catch (Exception e){
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        if(iMqttDeliveryToken.getException()!=null){
            log.error("MQPark##发布消息结束："+iMqttDeliveryToken.isComplete()+",发送异常：",iMqttDeliveryToken.getException());
        }
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("MQPark##已连接");
        MqttClient mqttClient = MqttParkClient.getMqttClient(MqttParkClient.MQTT_SERVER,
        		MqttParkClient.MQTT_PORT, MqttParkClient.MQTT_CLIENTID);
        List<ZsyyMqttConfig> mqttConfigList =mqttConfigMapper.selectList(null);
        List<String> topics = new ArrayList<>();
//        if (!CommonUtil.isEmpty(mqttConfigList)){
//            for (MqttConfig mqttConfig:mqttConfigList) {
//                topics.add(mqttConfig.getSubTopic());
//            }
//        }

        try {
            log.info("MQPark##当前clientID:{}", mqttClient.getClientId());
            for (int i = 0; i < topics.size(); i++) {
                mqttClient.subscribe(topics.get(i), 1);
                log.info("MQPark##订阅主题：{{}}",topics.get(i));
            }

        } catch (Exception e) {
            log.error("MQPark##连接回调异常", e);
        }
    }

}
